package xkafka

import (
	"sync"
	"time"

	"github.com/segmentio/kafka-go"
	"github.com/segmentio/kafka-go/sasl/plain"
	"gitlab-tech.shruisong.cn/growth/go-common/pkg/utils"
	"go.uber.org/zap"
)

type WriterConfig struct {
	Username     string
	Password     string
	Brokers      []string
	Topic        string
	ReadTimeout  time.Duration
	WriteTimeout time.Duration
	Async        bool
	Logger       Logger
}

type Writer struct {
	*kafka.Writer
	Name string
}

func NewProducer(name string, config WriterConfig) (writer *Writer, err error) {
	mKey := name + ":" + config.Topic
	if len(name) > 0 && pp.get(mKey) != nil {
		return pp.get(mKey), nil
	}

	if config.Logger == nil {
		logger := &DefaultLogger{}
		logger.SetPrefix(WriterLogPrefix)
		config.Logger = logger
	}
	writerConfig := kafka.WriterConfig{
		Brokers:      config.Brokers,
		Topic:        config.Topic,
		ReadTimeout:  config.ReadTimeout,
		WriteTimeout: config.WriteTimeout,
		Async:        config.Async,
		Logger:       kafka.LoggerFunc(config.Logger.Infof),
		ErrorLogger:  kafka.LoggerFunc(config.Logger.Errorf),
	}
	// 验证
	if len(config.Username) > 0 {
		dialer := &kafka.Dialer{
			SASLMechanism: plain.Mechanism{
				Username: config.Username,
				Password: config.Password,
			},
		}
		writerConfig.Dialer = dialer
	}

	err = writerConfig.Validate()
	if err != nil {
		config.Logger.Errorf("WriterConfig err: " + err.Error())
		return nil, err
	}
	//log.Println("new_kafka_producer")
	utils.Log().Debug("new kafka producer", zap.String("producer_key", mKey))
	kwriter := kafka.NewWriter(writerConfig)
	writer = &Writer{
		Name:   name,
		Writer: kwriter,
	}
	if len(name) > 0 {
		pp.add(mKey, writer)
	}
	return writer, nil
}

func CloseProducer(producer *Writer) error {
	if len(producer.Name) > 0 {
		pp.del(producer.Name)
	}

	return producer.Close()
}

func CloseAllProducers() {
	pp.clear()
}

var pp *producers

func init() {
	pp = &producers{
		producers: make(map[string]*Writer),
	}
}

type producers struct {
	producers map[string]*Writer
	sync.RWMutex
}

func (p *producers) add(name string, producer *Writer) {
	p.Lock()
	defer p.Unlock()
	p.producers[name] = producer
}

func (p *producers) get(name string) *Writer {
	p.RLock()
	defer p.RUnlock()
	return p.producers[name]
}

func (p *producers) del(name string) {
	p.Lock()
	defer p.Unlock()
	producer := p.producers[name]
	delete(p.producers, name)
	if producer != nil{
		producer.Close()
	}
}

func (p *producers) clear() {
	// p.Lock()
	// defer p.Unlock()
	for k := range p.producers {
		p.del(k)
	}
}

//func GetProducer
